import asyncio

from app.utils.kafka import KafkaConsumer, KafkaProducer
from app.search_service.teacher_service import TeacherService
from app.search_service.other_service import OtherService
from app.utils.logger import logger


def _is_valid_group_message(data):
    """
    检查是否为有效的群消息
    :param data: 消息内容
    :return: bool
    """
    return (
            data.get("message_type") == "group"
            and isinstance(data.get("raw_message"), str)
            and isinstance(data.get("group_id"), int)
    )


def _is_teacher_query(raw_message):
    """
    检查是否为老师查询
    :param raw_message: 原始消息内容
    :return: bool
    """
    if "老师" not in raw_message:
        return False

    index = raw_message.index("老师")
    # 检查“老师”前有字符且字符数在6个以内，同时“老师”后无字符
    return (
            0 < index <= 6  # “老师”前字符数在6个以内
            and index + len("老师") == len(raw_message)  # “老师”后无字符
    )


class MessageProcessor:
    def __init__(self):
        self.kafka_consumer = KafkaConsumer("qq_messages")  # 消费消息的主题
        self.kafka_producer = KafkaProducer()  # 发送消息的生产者
        self.teacher_service = TeacherService()
        self.other_service = OtherService()
        self.semaphore = asyncio.Semaphore(10)  # 最多10个并发任务

    async def start(self):
        """启动消息处理服务"""
        await self.kafka_producer.start()  # 在程序启动时初始化 Producer
        logger.info("Message_Processor start")
        async for msg in self.kafka_consumer.consume():
            asyncio.create_task(self._handle_with_limit(msg))

    async def stop(self):
        await self.kafka_producer.stop()  # 程序关闭时关闭 Producer

    async def _handle_with_limit(self, msg):
        async with self.semaphore:
            await self.process_message(msg)

    async def process_message(self, data):
        """
        处理从Kafka消费的消息
        :param data: 消息内容，包含 raw_message 和 group_id
        """

        logger.info(f"Received Kafka message: {data}")

        try:
            # 检查消息类型和字段
            if not _is_valid_group_message(data):
                return

            raw_message = data["raw_message"]
            group_id = data["group_id"]

            # 判断是否为老师查询
            if _is_teacher_query(raw_message):
                response = await self.teacher_service.handle(data)
                contain_type = 'text'
            else:
                response, contain_type = await self.other_service.handle(data)

            # 将响应消息发送到Kafka
            await self._send_response(group_id, response, contain_type)
        except Exception as e:
            logger.error(f"Error processing message: {e}")

    async def _send_response(self, group_id, message, contain_type):
        """
        将响应消息发送到Kafka
        :param group_id: 群组ID
        :param message: 响应消息
        """
        try:
            await self.kafka_producer.send("qq_send_messages", {
                "group_id": group_id,
                "message": message,
                "contain_type": contain_type
            })
            logger.info(f"Response message sent to Kafka for group {group_id}")
        except Exception as e:
            logger.error(f"Failed to send response message to Kafka: {e}")
